/*
 * Copyright 2009 Armando Blancas
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package taskgraph.tasks.data;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Collection;
import java.util.Iterator;
import java.util.NoSuchElementException;

import taskgraph.ports.OutputPort;
import taskgraph.tasks.PrimitiveTask;


/**
 * Reads a collection, putting each element through the output port.
 * 
 * <p>Elements read from the collection may optionally be copied
 * (or cloned) in order to ensure the safety of concurrent access.
 * If this is requested, elements must provide a copy constructor
 * or clone() method.
 * 
 * @author Armando Blancas
 * @see taskgraph.ports.OutputPort
 * @see java.util.Collection
 * @param <E> The type of elements transported by this task.
 */
public class ReadCollection<E> extends PrimitiveTask {

	private OutputPort<E> output;
	private Collection<? extends E> collection;
	private boolean duplicating; 
	
	/**
	 * This constructor allows creating instances as beans.
	 */
    public ReadCollection() {
    	// do nothing
    }

	/**
	 * Constructor.
     * 
     * @param output The output port.
     * @param collection A collection with the source data.
     */
    public ReadCollection(final OutputPort<E> output,
    		              final Collection<? extends E> collection) {
    	setOutput(output);
    	setCollection(collection);
    }

    /**
	 * Gets the collection that provides the source data.
	 * 
	 * @return The collection reference.
	 */
	public Collection<? extends E> getCollection() {
		return collection;
	}

	/**
     * Gets the output port.
     * 
	 * @return The output port.
	 */
	public OutputPort<E> getOutput() {
		return output;
	}

	/**
	 * Tests whether the task is making element duplicates.
	 * 
	 * @return The {@code boolean} value.
	 */
	public boolean isDuplicating() {
		return duplicating;
	}

	/**
     * Performs the work of this task.
     * 
	 * Reads elements from the collection and writes them to the output port. 
	 *<pre>
	 *If task is duplicating elements
	 *    if class E has a copy constructor
	 *        loop through input
	 *        copy element
	 *        write copy to output port
	 *    if class E defines clone()
	 *        loop through input
	 *        clone element
	 *        write clone to output port
	 *else
	 *    while not EOF
	 *        read an object reference from the collection
	 *        write it to the output port
	 *close output port
	 *</pre>
	 */
	@SuppressWarnings("unchecked")
	@Override
    public void run() {
    	try {
    		Iterator<? extends E> itor = collection.iterator();
    		if (isDuplicating()) {
				E element = itor.next();
    			Class<?> clazz = element.getClass();
    			Constructor constructor = clazz.getDeclaredConstructor(clazz);    			
    			if (constructor != null) {
    	    		while (!Thread.currentThread().isInterrupted()) {
    	    			output.write((E) constructor.newInstance(element));
    	    			element = itor.next();
    	    		}
    			} else {
    				Method clone = clazz.getDeclaredMethod("clone");
    				while (!Thread.currentThread().isInterrupted()) {
            			output.write((E) clone.invoke(element));    				
            			element = itor.next();
    				}
    			}
    		} else { // Default: not duplicating:
    			while (!Thread.currentThread().isInterrupted()) {
    				E element = itor.next();
    				output.write(element);
    			}
    		}
        } catch (InterruptedIOException i) {
        	// interruption request
        } catch (NoSuchElementException e) {
        	// we're done reading the collection
		} catch (SecurityException e) {
            log.error("ReadCollection#run()", e);
		} catch (NoSuchMethodException e) {
            log.error("ReadCollection#run()", e);
		} catch (IllegalArgumentException e) {
            log.error("ReadCollection#run()", e);
		} catch (InstantiationException e) {
            log.error("ReadCollection#run()", e);
		} catch (IllegalAccessException e) {
            log.error("ReadCollection#run()", e);
		} catch (InvocationTargetException e) {
            log.error("ReadCollection#run()", e);
        } catch (IOException e) {
        	// pipe synch error
            log.error("ReadCollection#run()", e);
		} finally {
			output.close();
        }
    }

	/**
	 * Sets the collection that provides the source data.
	 * 
	 * @param collection The collection reference to set.
	 */
	public void setCollection(Collection<? extends E> collection) {
    	if (collection == null)
    		throw new IllegalArgumentException("collection == null");
		this.collection = collection;
	}

	/**
	 * Sets whether to make element duplicates.
	 * 
	 * @param duplicating The {@code boolean} value to set.
	 */
	public void setDuplicating(boolean duplicating) {
		this.duplicating = duplicating;
	}

	/**
	 * Sets the output port.
	 * 
	 * @param output The output port.
	 */
    public void setOutput(final OutputPort<E> output) {
    	if (output == null)
    		throw new IllegalArgumentException("output == null");
 		this.output = output;
	}

}
